package com.gitee.Jmysy.binlog4j.core;

import com.gitee.Jmysy.binlog4j.core.config.RedisConfig;
import com.gitee.Jmysy.binlog4j.core.dispatcher.BinlogEventDispatcher;
import com.gitee.Jmysy.binlog4j.core.enums.BinlogClientMode;
import com.gitee.Jmysy.binlog4j.core.exception.Binlog4jException;
import com.gitee.Jmysy.binlog4j.core.position.BinlogPosition;
import com.gitee.Jmysy.binlog4j.core.position.BinlogPositionHandler;
import com.gitee.Jmysy.binlog4j.core.position.RedisBinlogPositionHandler;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * Binlog 客户端
 *
 * @author 就眠儀式
 * */
@Slf4j
public class BinlogClient implements IBinlogClient {

    private final BinlogClientConfig clientConfig;

    private BinaryLogClient client;

    private BinlogPositionHandler positionHandler;

    private RedissonClient redissonClient;

    private final Map<String, BinlogEventHandlerInvoker> eventHandlerList = new HashMap<>();

    private final ExecutorService executor;

    public BinlogClient(BinlogClientConfig clientConfig) {
        this.createPositionHandler(clientConfig);
        this.createRedissonClient(clientConfig);
        this.clientConfig = clientConfig;
        this.executor = Executors.newCachedThreadPool();
    }

    @Override
    public void registerEventHandler(String handlerKey, IBinlogEventHandler eventHandler) {
        BinlogEventHandlerInvoker eventHandlerDetails = new BinlogEventHandlerInvoker();
        eventHandlerDetails.setClientConfig(clientConfig);
        eventHandlerDetails.setEventHandler(eventHandler);
        this.eventHandlerList.put(handlerKey, eventHandlerDetails);
    }

    @Override
    public void registerEventHandler(IBinlogEventHandler eventHandler) {
        this.registerEventHandler(UUID.randomUUID().toString(), eventHandler);
    }

    @Override
    public void unregisterEventHandler(String handlerKey) {
        if(eventHandlerList.containsKey(handlerKey)) {
            eventHandlerList.remove(handlerKey);
        }
    }

    @Override
    public void connect() {
        BinlogClientMode clientMode = clientConfig.getMode();
        if (clientMode == BinlogClientMode.cluster) {
            ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(1);
            scheduledExecutor.scheduleWithFixedDelay(this::runWithCluster, 0, 1000, TimeUnit.MILLISECONDS);
        } else {
            executor.submit(this::runWithStandalone);
        }
    }

    @Override
    public void disconnect() {
        if (client != null) {
            try {
                client.disconnect();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public void runWithStandalone() {
        try {
            log.info("启动 Binlog 客户端 ({}) - 连接 {}:{} 服务", clientConfig.getServerId(), clientConfig.getHost(), clientConfig.getPort());
            client = new BinaryLogClient(clientConfig.getHost(), clientConfig.getPort(), clientConfig.getUsername(), clientConfig.getPassword());
            client.registerEventListener(new BinlogEventDispatcher(this.clientConfig, positionHandler, this.eventHandlerList));
            client.setKeepAlive(clientConfig.getKeepAlive());
            client.setKeepAliveInterval(clientConfig.getKeepAliveInterval());
            client.setHeartbeatInterval(clientConfig.getHeartbeatInterval());
            client.setConnectTimeout(clientConfig.getConnectTimeout());
            client.setServerId(clientConfig.getServerId());
            if (clientConfig.getPersistence()) {
                if (!clientConfig.isInaugural()) {
                    if (positionHandler != null) {
                        BinlogPosition binlogPosition = positionHandler.loadPosition(clientConfig.getServerId());
                        if (binlogPosition != null) {
                            client.setBinlogFilename(binlogPosition.getFilename());
                            client.setBinlogPosition(binlogPosition.getPosition());
                        }
                    }
                }
            }
            client.connect();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void runWithCluster() {
        RLock lock = redissonClient.getLock(clientConfig.getKey());
        try {
            if (lock.tryLock()) {
                runWithStandalone();
            }
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

    private void createPositionHandler(BinlogClientConfig clientConfig) {
        if(clientConfig.getPersistence()) {
            if(clientConfig.getPositionHandler() == null) {
                if(clientConfig.getRedisConfig() == null) {
                    throw new Binlog4jException("Cluster mode or persistence enabled, missing Redis configuration");
                } else {
                    this.positionHandler = new RedisBinlogPositionHandler(clientConfig.getRedisConfig());
                }
            } else {
                this.positionHandler = clientConfig.getPositionHandler();
            }
        }
    }

    private void createRedissonClient(BinlogClientConfig clientConfig) {
        if (clientConfig.getMode() == BinlogClientMode.cluster) {
            RedisConfig redisConfig = clientConfig.getRedisConfig();
            if (redisConfig == null) {
                throw new Binlog4jException("Cluster mode or persistence enabled, missing Redis configuration");
            }
            Config config = new Config();
            SingleServerConfig singleServerConfig = config.useSingleServer();
            singleServerConfig.setAddress("redis://" + redisConfig.getHost() + ":" + redisConfig.getPort());
            singleServerConfig.setPassword(redisConfig.getPassword());
            singleServerConfig.setDatabase(redisConfig.getDatabase());
            config.setLockWatchdogTimeout(10000L);
            this.redissonClient = Redisson.create(config);
        }
    }
}
